package org.latestbit.smartobjects.perftests.client;

import java.util.concurrent.atomic.AtomicInteger;

import org.latestbit.smartobjects.ISmartBus;
import org.latestbit.smartobjects.perftests.IPerfTestService;

public class SyncPerfClientImpl extends AbstractPerfTestImpl implements Runnable {
	AtomicInteger tps = new AtomicInteger(0);
	private Thread statThread;

	public SyncPerfClientImpl(
			ISmartBus bus, 
			CommandLineOptions options,
			IPerfTestService service) {
		super(bus, options, service);
		this.statThread = new Thread(new StatThread());
	}

	@Override
	public void start() throws Exception {
		statThread.start();
		for(int i=0; i< options.getThreadsCount(); i++)
			threadPool.execute(this);
	}

	@Override
	public void stop() {
		stopFlag = true;
		try {
			statThread.join();
		} 
		catch (Exception e) {
			e.printStackTrace();
		}
		threadPool.shutdown();		
	}

	@Override
	public void run() {
		while(!stopFlag) {
			try {
				this.service.testMethod(
						createTestParamStr(), 
						createTestParamInt(), 
						createTestStructure(), 
						createTestStructureArray(this.options.getArraySize())
				);
				tps.addAndGet(1);
				if(options.getDelay()!=null) {
					Thread.sleep(options.getDelay());
				}
			} 
			catch (Exception ex) {
				ex.printStackTrace();
			}
		}
	}

	protected class StatThread implements Runnable {

		@Override
		public void run() {
			while(!stopFlag) {
				try {
					Thread.sleep(1000);
					
					int lTps = tps.getAndSet(0);
					long bytes = lTps * packetSize;
					long kbit = (bytes/1024) * 8;
					long mbit = (bytes/1024/1024)*8;
					
					System.out.println("TPS: "+ lTps+" Bytes: "+bytes
							+" "+(mbit <=1 ? kbit+" Kbit" : mbit+" Mbit"  )
					);
				} 
				catch (Exception e) {
					e.printStackTrace();
				}				
			}
		}
		
	}
}
